package com.whq.thrift.api.common.thriftclient;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.thrift.async.TAsyncClientManager;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.protocol.TProtocolFactory;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TNonblockingTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransportException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.whq.common.CollectionUtil;
import com.whq.common.StringUtil;
import com.whq.common.UUIDUtil;
import com.whq.common.http.HttpClientUtil;
import com.whq.thrift.api.common.constants.CommonConfigConstants;
import com.whq.thrift.api.common.constants.CommonErrorCodeConstants;
import com.whq.thrift.api.common.thriftclient.model.IpPortInfo;
import com.whq.thrift.api.common.thriftclient.model.ResponseModuleList;
import com.whq.thrift.api.common.thriftclient.model.ServiceModuleInfo;
import com.whq.thrift.api.common.thriftserver.ApiRequest;
import com.whq.thrift.api.common.thriftserver.ResponseBase;

public class ThriftApiJsonClientHelper {
	private static final Logger LOGGER = LoggerFactory.getLogger(ThriftApiJsonClientHelper.class);
	
	private static final Logger LOGGER_QUARTZ = LoggerFactory.getLogger(CommonConfigConstants.LOGGER_KEY_QUARTZ);
	private static final Logger LOGGER_THRIFT_REQUEST_RESPONSE = LoggerFactory.getLogger(CommonConfigConstants.LOGGER_KEY_THRIFT_REQUEST_RESPONSE);
	
	// Thrift服务标识 => Thrift信息
	protected static final Map<String, List<ThriftServiceRecord>> MAP_THRIFT_SERVICE_CONFIG = new ConcurrentHashMap<String, List<ThriftServiceRecord>>(0);
	
	// Thrift服务标识 => 上次调用的Thrift实例的位置 （用来处理负载，使用轮询机制）
	private static final Map<String, Integer> MAP_THRIFT_NEXT_INVOKE_INDEX = new ConcurrentHashMap<String, Integer>(0);
	
	private static TAsyncClientManager ASYNC_CLIENT_MANAGER = null;
	private static TProtocolFactory PROTOCAL_FACTORY = null;
	
	// 例子：http://192.168.18.78:6200/apipool?method=base_service_config.getServiceInstance
	private String thriftConfigUrl = "";
	
	// 例子：[{"moduleId":"txz_bs_gesture","instanceList":[{"port":8112,"ip":"10.240.17.86"}]}]
	private String thriftServicesConfigJson = "";
	
	/**
	 * 初始化
	 */
	public void initialize() {
		if (ASYNC_CLIENT_MANAGER == null) {
			try {
				ASYNC_CLIENT_MANAGER = new TAsyncClientManager();
			} catch (IOException e) {
				LOGGER.error("new TAsyncClientManager() exception: ", e); 
			}
		}
		
		if (PROTOCAL_FACTORY == null) {
			PROTOCAL_FACTORY = new TBinaryProtocol.Factory(); 
		}
		
		if (StringUtil.isNotBlank(thriftServicesConfigJson)) {
			List<ServiceModuleInfo> moduleList = null;
			try {
				moduleList = JSON.parseArray(thriftServicesConfigJson, ServiceModuleInfo.class); 
			} catch (Exception e) {
				LOGGER.error("parse response[" + thriftServicesConfigJson + "] to List<ServiceModuleInfo> exception: ", e);
			}
			
			if (moduleList != null) {
				refreshServiceInstance(moduleList);	
			}
		}
	}
	
	public void shutDown() {
		
	}
	
	/**
	 * 请求url刷新Thrift服务实例
	 */
	public void refreshAllServiceInstancesFromUrl() {
		LOGGER_QUARTZ.info("begin refresh ice service client instances from url: " + thriftConfigUrl);
		if (StringUtil.isBlank(thriftConfigUrl)) {
			return ;
		}
		
		Map<String, String> paramsMap = new HashMap<String, String>(0);
		paramsMap.put("moduleIds", CollectionUtil.convertSet2String(MAP_THRIFT_SERVICE_CONFIG.keySet(), ","));
		
		String responseStr = HttpClientUtil.fetchStrByGetMethod(thriftConfigUrl, paramsMap, false, 3000, 3000);
		
		LOGGER_QUARTZ.info("request url[" + thriftConfigUrl +"], params[" + paramsMap + "], response[" + responseStr + "]");
		ResponseModuleList responseModuleList = null;
		try {
			responseModuleList = JSON.parseObject(responseStr, ResponseModuleList.class);
		} catch (Exception e) {
			LOGGER.error("parse response[" + responseStr + "] to ResponseModuleList exception: ", e);
		}
		
		if ( (responseModuleList != null)
				&& (responseModuleList.getReturnCode() == CommonErrorCodeConstants.ERR_CODE_SUCCESS)
				&& (responseModuleList.getData() != null)
				){
			refreshServiceInstance(responseModuleList.getData());	
		}
		
		LOGGER_QUARTZ.info("end refresh ice service client instances from url: " + thriftConfigUrl);
	}
	
	private String getKeyByIpPort(String ip, int port) {
		return ip + "-" + port;
	}
	
	/**
	 * 刷新配置实例
	 * @param serviceModuleList
	 */
	protected void refreshServiceInstance(List<ServiceModuleInfo> serviceModuleList) {
		if (serviceModuleList == null) {
			return ;
		}
		
		for (ServiceModuleInfo serviceModuleInfo : serviceModuleList) {
			if (serviceModuleInfo == null) {
				continue ;
			}
			
			Set<String> newServiceRecordKeySet = new HashSet<String>(0);
			
			List<IpPortInfo> newInstanceList = serviceModuleInfo.getInstanceList();
			if (newInstanceList != null) {
				for (IpPortInfo ipPortInfo : newInstanceList) {
					if (ipPortInfo == null) {
						continue;
					}
					newServiceRecordKeySet.add(getKeyByIpPort(ipPortInfo.getIp(), ipPortInfo.getPort()));
				}
			}
			
			String thriftServiceIdentity = serviceModuleInfo.getModuleId();
			Set<String> oldServiceRecordKeySet = new HashSet<String>(0);
			// 先用的所有服务改成待清理状态
			List<ThriftServiceRecord> thriftServiceRecordList = getThriftServiceRecordFromMap(thriftServiceIdentity);
			if (thriftServiceRecordList == null) {
				thriftServiceRecordList = new ArrayList<ThriftServiceRecord>(0);
			}
			for (Iterator<ThriftServiceRecord> iterator = thriftServiceRecordList.iterator(); iterator.hasNext();) {
				ThriftServiceRecord existServiceRecord = (ThriftServiceRecord) iterator.next();
				if (existServiceRecord != null) {
					String ipPortKey = getKeyByIpPort(existServiceRecord.getIp(), existServiceRecord.getPort());
					
					oldServiceRecordKeySet.add(ipPortKey);
					
					if (!newServiceRecordKeySet.contains(ipPortKey)) {
						iterator.remove();
					}
				}
			}
			
			// 添加新的实例
			if (newInstanceList != null) {
				for (IpPortInfo ipPortInfo : newInstanceList) {
					if (ipPortInfo == null) {
						continue;
					}
					String ipPortKey = getKeyByIpPort(ipPortInfo.getIp(), ipPortInfo.getPort());
					
					if (!oldServiceRecordKeySet.contains(ipPortKey)) {
						ThriftServiceRecord thriftServiceRecord = new ThriftServiceRecord();
						thriftServiceRecord.setIp(ipPortInfo.getIp());
						thriftServiceRecord.setPort(ipPortInfo.getPort());
						thriftServiceRecord.setValid(true);
						
						thriftServiceRecordList.add(thriftServiceRecord);
					}
				}
			}
			
			MAP_THRIFT_SERVICE_CONFIG.put(thriftServiceIdentity, thriftServiceRecordList);
		}
	}
	
	/**
	 * 通过ping的方式检查所有实例是否正常
	 */
	public void checkAllServiceValid() {
		LOGGER_QUARTZ.info("begin check ice service client instance valid");
		Set<String> serviceModuleIdSet = MAP_THRIFT_SERVICE_CONFIG.keySet();
		for (Iterator<String> iterator = serviceModuleIdSet.iterator(); iterator.hasNext();) {
			final String thriftServiceIdentity = (String) iterator.next();
			List<ThriftServiceRecord> thriftServiceRecordList = getThriftServiceRecordFromMap(thriftServiceIdentity);
			if ( (thriftServiceRecordList == null)
					|| (thriftServiceRecordList.isEmpty()) 
					){
				continue;
			}
			
			LOGGER_QUARTZ.info("begin check ice service client instance: " + thriftServiceIdentity);
			
			for (Iterator<ThriftServiceRecord> iterator2 = thriftServiceRecordList.iterator(); iterator2.hasNext();) {
				final ThriftServiceRecord thriftServiceRecord = (ThriftServiceRecord) iterator2.next();
				if (thriftServiceRecord == null) {
					continue;
				}
				
				LOGGER_QUARTZ.info("begin check ice service client instance: " + thriftServiceIdentity + ", ip[" + thriftServiceRecord.getIp() + "], port[" + thriftServiceRecord.getPort() + "]");
				
				try { 
					TFramedTransport transport = new TFramedTransport(new TSocket(thriftServiceRecord.getIp(), thriftServiceRecord.getPort(), 1000)); 
					transport.open();
					TProtocol protocol = new TBinaryProtocol(transport); 
		            ApiRequest.Client syncClient = new ApiRequest.Client(protocol); 
		            
		            Map<String, String> params = new HashMap<String, String>(0);
		            params.put(CommonConfigConstants.THRIFT_PARAM_METHOD, CommonConfigConstants.THRIFT_PARAM_METHOD_VAL_KEEP_ALIVE);
		            Map<String, String> ctx = new HashMap<String, String>(0);
		            
		            String responseStr = syncClient.handleRequest(params, ctx);
//		            transport.close();
		            
		            LOGGER_QUARTZ.info("invoke method[" + CommonConfigConstants.THRIFT_PARAM_METHOD_VAL_KEEP_ALIVE + "], ip[" + thriftServiceRecord.getIp() + "], port[" + thriftServiceRecord.getPort() + "] response[" + responseStr + "]");
		            
		            ResponseBase responseBase = JSON.parseObject(responseStr, ResponseModuleList.class);
		            if ( (responseBase != null) 
		            		&& (responseBase.getReturnCode() == CommonErrorCodeConstants.ERR_CODE_SUCCESS) 
		            		) {
		            	thriftServiceRecord.setValid(true);
		            	LOGGER_QUARTZ.info("ice service client instance: " + thriftServiceIdentity + ", ip[" + thriftServiceRecord.getIp() + "], port[" + thriftServiceRecord.getPort() + "] is valid");
		            } else {
		            	thriftServiceRecord.setValid(false);
		            	LOGGER_QUARTZ.info("ice service client instance: " + thriftServiceIdentity + ", ip[" + thriftServiceRecord.getIp() + "], port[" + thriftServiceRecord.getPort() + "] is invalid");
		            }
		        } catch (Exception e) { 
		        	thriftServiceRecord.setValid(false);
		        	
		            e.printStackTrace();
		            LOGGER_QUARTZ.info("ice service client instance: " + thriftServiceIdentity + ", ip[" + thriftServiceRecord.getIp() + "], port[" + thriftServiceRecord.getPort() + "] is invalid");
		            LOGGER.error("check ice service client instance: " + thriftServiceIdentity + ", ip[" + thriftServiceRecord.getIp() + "], port[" + thriftServiceRecord.getPort() + "] exception: ", e);
		        }
				
				LOGGER_QUARTZ.info("end check ice service client instance: " + thriftServiceIdentity + ", ip[" + thriftServiceRecord.getIp() + "], port[" + thriftServiceRecord.getPort() + "]");
			}
			
			LOGGER_QUARTZ.info("end check ice service client instance: " + thriftServiceIdentity);
		}
		LOGGER_QUARTZ.info("end check ice service client instance valid");
	}
	
	/**
	 * 调用thrift服务(同步)
	 * @param thriftServiceIdentity
	 * @param parameters
	 * @param ctx
	 * @param invokeTimeoutMs
	 * @param fromMethod
	 * @return
	 */
	public String syncInvokeThriftService(String thriftServiceIdentity, Map<String, String> parameters, Map<String, String> ctx, 
			int invokeTimeoutMs, String fromMethod) {
		LOGGER_THRIFT_REQUEST_RESPONSE.info("sync invoke thrift service fromMethod[" + fromMethod + "], params[" + parameters + "], ctx[" + ctx + "] begin");
		
		long startMs = System.currentTimeMillis();
		
		String method = parameters.get(CommonConfigConstants.THRIFT_PARAM_METHOD);
		
		String flowId = ctx.get(CommonConfigConstants.THRIFT_CTX_FLOW_ID);
		if (StringUtil.isBlank(flowId)) {
			flowId = UUIDUtil.generateUUID();
			ctx.put(CommonConfigConstants.THRIFT_CTX_FLOW_ID, flowId);
		}
		
		ThriftServiceRecord thriftServiceRecord = getThriftServiceInstance(thriftServiceIdentity, invokeTimeoutMs);
		TFramedTransport transport = null;
		if (thriftServiceRecord != null) {
			transport = thriftServiceRecord.getFramedTransport();
			if (transport == null) {
				thriftServiceRecord.initFramedTransport();
				transport = thriftServiceRecord.getFramedTransport();
			}
			if (transport != null) {
				if (!transport.isOpen()) {
					try {
						transport.open();
					} catch (TTransportException e) {
						e.printStackTrace();
					}
				}
				if (!transport.isOpen()) {
					thriftServiceRecord.initFramedTransport();
					transport = thriftServiceRecord.getFramedTransport();
					if (transport != null) {
						try {
							transport.open();
						} catch (TTransportException e) {
							e.printStackTrace();
						}
					}
				}
			}
		}
		
		long getInstanceConsumeMs = System.currentTimeMillis() - startMs;
		
		LOGGER_THRIFT_REQUEST_RESPONSE.info("sync invoke thrift service fromMethod[" + fromMethod + "],  params[" + parameters + "], ctx[" + ctx + "] get ice client instance consume " + getInstanceConsumeMs + " ms");
		
		if (transport == null) {
			String ret = JSON.toJSONString(new ResponseBase(CommonErrorCodeConstants.ERR_CODE_THRIFT_INSTANCE_NULL, "Thrift服务实例[" + thriftServiceIdentity + "]不存在"));
			long timeConsumeMs = System.currentTimeMillis() - startMs;
			
			String ip = "";
			String port = "";
			if (thriftServiceRecord != null) {
				ip = thriftServiceRecord.getIp();
				port = "" + thriftServiceRecord.getPort();
			}
			
			LOGGER_THRIFT_REQUEST_RESPONSE.info("sync invoke thrift service fromMethod[" + fromMethod + "],  instance[" + ip + ", " + port + "], params[" + parameters + "], ctx[" + ctx + "], response[" + ret + "], consume " + timeConsumeMs + " ms");
			return ret;
		}
		
		LOGGER_THRIFT_REQUEST_RESPONSE.info("sync invoke thrift service fromMethod[" + fromMethod + "],  instance[" + thriftServiceRecord.getIp() + ", " + thriftServiceRecord.getPort() + "], params[" + parameters + "], ctx[" + ctx + "] begin invoke");
		
		String response = JSON.toJSONString(new ResponseBase(CommonErrorCodeConstants.ERR_CODE_THRIFT_RESPONSE_NULL, "Thrift服务[" + method + "]没有响应"));;
		try {
			TProtocol protocol = new TBinaryProtocol(transport); 
	        ApiRequest.Client syncClient = new ApiRequest.Client(protocol); 
	        response = syncClient.handleRequest(parameters, ctx);
		} catch (Exception e) {
			LOGGER.error("sync invoke thrift service fromMethod[" + fromMethod + "],  instance[" + thriftServiceRecord.getIp() + ", " + thriftServiceRecord.getPort() + "], params[" + parameters + "], ctx[" + ctx + "] exception: ", e);
		}
		
		long timeConsumeMs = System.currentTimeMillis() - startMs;
		LOGGER_THRIFT_REQUEST_RESPONSE.info("sync invoke thrift service method[" + method + "], flowId[" + flowId + "], fromMethod[" + fromMethod + "], response[" + response + "], consume " + timeConsumeMs + " ms");
		
		return response;
	}
	
	/**
	 * 调用thrift服务(异步)
	 * @param callback
	 * @param thriftServiceIdentity
	 * @param parameters
	 * @param ctx
	 * @param invokeTimeoutMs
	 * @param fromMethod
	 */
	public void asyncInvokeThriftService(AbstractApiJsonCallback callback, String thriftServiceIdentity, Map<String, String> parameters, Map<String, String> ctx, 
			int invokeTimeoutMs, String fromMethod) {
		if (callback == null) {
			return ;
		}
		
		LOGGER_THRIFT_REQUEST_RESPONSE.info("invoke thrift service fromMethod[" + fromMethod + "], params[" + parameters + "], ctx[" + ctx + "] begin");
		
		long startMs = System.currentTimeMillis();
		callback.setStartMs(startMs);
		
		String method = parameters.get(CommonConfigConstants.THRIFT_PARAM_METHOD);
		callback.setMethod(method);
		
		String flowId = ctx.get(CommonConfigConstants.THRIFT_CTX_FLOW_ID);
		if (StringUtil.isBlank(flowId)) {
			flowId = UUIDUtil.generateUUID();
			ctx.put(CommonConfigConstants.THRIFT_CTX_FLOW_ID, flowId);
		}
		callback.setFlowId(flowId);
		
		ThriftServiceRecord thriftServiceRecord = getThriftServiceInstance(thriftServiceIdentity, invokeTimeoutMs);
		TNonblockingTransport transport = null;
		if (thriftServiceRecord != null) {
			transport = thriftServiceRecord.getNonblockingTransport();
			if (transport == null) {
				thriftServiceRecord.initNonblockingTransport();
				transport = thriftServiceRecord.getNonblockingTransport();
			}
			if (transport != null) {
				if (!transport.isOpen()) {
					thriftServiceRecord.initNonblockingTransport();
					transport = thriftServiceRecord.getNonblockingTransport();
				}
			}
		}
		
		long getInstanceConsumeMs = System.currentTimeMillis() - startMs;
		
		LOGGER_THRIFT_REQUEST_RESPONSE.info("invoke thrift service fromMethod[" + fromMethod + "],  params[" + parameters + "], ctx[" + ctx + "] get ice client instance consume " + getInstanceConsumeMs + " ms");
		
		if (transport == null) {
			String ret = JSON.toJSONString(new ResponseBase(CommonErrorCodeConstants.ERR_CODE_THRIFT_INSTANCE_NULL, "Thrift服务实例[" + thriftServiceIdentity + "]不存在"));
			long timeConsumeMs = System.currentTimeMillis() - startMs;
			
			String ip = "";
			String port = "";
			if (thriftServiceRecord != null) {
				ip = thriftServiceRecord.getIp();
				port = "" + thriftServiceRecord.getPort();
			}
			
			LOGGER_THRIFT_REQUEST_RESPONSE.info("invoke thrift service fromMethod[" + fromMethod + "],  instance[" + ip + ", " + port + "], params[" + parameters + "], ctx[" + ctx + "], response[" + ret + "], consume " + timeConsumeMs + " ms");
			callback.doResponseApiJson(ret);
			return ;
		}
		
		LOGGER_THRIFT_REQUEST_RESPONSE.info("invoke thrift service fromMethod[" + fromMethod + "],  instance[" + thriftServiceRecord.getIp() + ", " + thriftServiceRecord.getPort() + "], params[" + parameters + "], ctx[" + ctx + "] begin invoke");
		
		ApiRequest.AsyncClient asyncClient = new ApiRequest.AsyncClient(PROTOCAL_FACTORY, ASYNC_CLIENT_MANAGER, transport);
		
		try {
			asyncClient.setTimeout(invokeTimeoutMs);
			asyncClient.handleRequest(parameters, ctx, callback);
		} catch (Exception e) {
			LOGGER.error("invoke thrift service fromMethod[" + fromMethod + "],  instance[" + thriftServiceRecord.getIp() + ", " + thriftServiceRecord.getPort() + "], params[" + parameters + "], ctx[" + ctx + "] exception: ", e);
		}
	}
	
	/**
	 * 根据方法名获取一个实例
	 * @param thriftServiceIdentity
	 * @return
	 */
	private ThriftServiceRecord getThriftServiceInstance(String thriftServiceIdentity, int invokeTimeoutMs) {
		List<ThriftServiceRecord> thriftServiceList = getThriftServiceRecordFromMap(thriftServiceIdentity);
		// 如果不存在，则访问url获取最新的实例
		if ( (thriftServiceList == null) 
				|| (thriftServiceList.isEmpty())
				){
			MAP_THRIFT_SERVICE_CONFIG.put(thriftServiceIdentity, new ArrayList<ThriftServiceRecord>(0));
			
			refreshAllServiceInstancesFromUrl();
			
			thriftServiceList = getThriftServiceRecordFromMap(thriftServiceIdentity);
			if ( (thriftServiceList == null) 
					|| (thriftServiceList.isEmpty())
					){
				return null;
			}
		}
		int nextInvokeIndex = 0;
		Integer invokeIndexFromCache = MAP_THRIFT_NEXT_INVOKE_INDEX.get(thriftServiceIdentity);
		if (invokeIndexFromCache != null) {
			nextInvokeIndex = invokeIndexFromCache.intValue();
		}
		
		int serviceCount = thriftServiceList.size();
		if (serviceCount <= 0) {
			return null;
		}
		
		int realIndex = nextInvokeIndex;
		for (int i = 0; i < serviceCount; i++ ) {
			if ( (realIndex < 0)
				 || (realIndex >= serviceCount)) {
				realIndex = 0;
			}
			
			ThriftServiceRecord thriftServiceRecord = null;
			try {
				thriftServiceRecord = thriftServiceList.get(realIndex);
			} catch (Exception e) {
				LOGGER.error("get ThriftServiceRecord[identity: " + thriftServiceIdentity + "][index: " + realIndex + "] from list [count: " + serviceCount + "] exception: ", e);
			}
			
			realIndex++;
			MAP_THRIFT_NEXT_INVOKE_INDEX.put(thriftServiceIdentity, realIndex);
			
			if (thriftServiceRecord == null) {
				continue;
			}
			
			if (!thriftServiceRecord.isValid()) {
				continue;
			}
			
			if (thriftServiceRecord.getNonblockingTransport() == null) {
				thriftServiceRecord.initNonblockingTransport();
			}
			
			if (thriftServiceRecord.getFramedTransport() == null) {
				thriftServiceRecord.initFramedTransport();
			}
			
			if (thriftServiceRecord.getNonblockingTransport() != null) {
				return thriftServiceRecord;
			}
		}
		return null;
	}
	
	protected static List<ThriftServiceRecord> getThriftServiceRecordFromMap(String thriftServiceIdentity) {
		List<ThriftServiceRecord> retList = new ArrayList<ThriftServiceRecord>(0);
		List<ThriftServiceRecord> thriftServiceList = MAP_THRIFT_SERVICE_CONFIG.get(thriftServiceIdentity);
		if (thriftServiceList != null) {
			retList.addAll(thriftServiceList);
		}
		return retList;
	}

	public String getThriftConfigUrl() {
		return thriftConfigUrl;
	}

	public void setThriftConfigUrl(String thriftConfigUrl) {
		this.thriftConfigUrl = thriftConfigUrl;
	}

	public String getThriftServicesConfigJson() {
		return thriftServicesConfigJson;
	}

	public void setThriftServicesConfigJson(String thriftServicesConfigJson) {
		this.thriftServicesConfigJson = thriftServicesConfigJson;
	}
	
}
